/**
The main program for the bulk loader. This file depends on the file
FileLoaderDefs.m4. In particular, it needs the following m4 macros defined:

	A. The datatype parameters
			RELATION_NAME: name of the relation (name of metadata and XML files)
			COLUMN_TYPES: description of the columns of the relation

	B. Storage parameters
			ASSUME: file diskArray.meta is in the current directory
			TUPLES_PER_CHUNK: number of tuples/chunk

	C. Input specification parameters
			NUM_INPUT_FRAGMENTS: number of fragments the text file is split into
			INPUT_FILE_PATTERN: pattern specifying the input file
													must contain %d if more than 1 fragment
													input file can be a named pipe  (text file not stored)

*/

dnl # M4 CODE
include(FileLoaderFunctions.m4)
m4_include(FileLoaderDefs.m4)
dnl # END OF M4 CODE

#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <iostream>
#include <iomanip>
#include <iostream>

#include "DataTypes.h"
#include "FileWriter.h"
#include "FileLoader.h"
#include "DistributedCounter.h"
#include "MetadataDB.h"
#include "DiskArray.h"
#include "Timer.h"

using namespace std;


/** this is a sample program on how to bulk load the database */

int main(int argc, char* argv[]){

	if (argc != 2) {
		cout << "Usage: BulkLoader [metadataDB file]" << endl;
		exit(-1);
	}

	Timer t;
	t.Restart();

	SetMetadataDB(argv[1]);
	DiskArray::LoadDiskArray();

	// initialize the dictionary for DString type
	DString::InitializeDictionary();

	int numInputFragments=NUM_INPUT_FRAGMENTS;
	int numDisks=NUM_DISKS;

	char* fileNames[numInputFragments];
	char fileNamesBuffer[10000];

	char* curr = fileNamesBuffer;
	for (int i=1; i<=numInputFragments; i++){
		fileNames[i-1]=curr;
		sprintf(curr, "INPUT_FILE_PATTERN", (i % numDisks) + 1, i);
		curr+=strlen(curr)+1;
	}

	// start the file writter
	FileWriter fileWriter("RELATION_NAME", M4_NUM_COLS(</COLUMN_TYPES/>));
	fileWriter.ForkAndSpin();

	DistributedCounter* dCounter = new DistributedCounter(numInputFragments);

	// start the generators
	TwoWayList <EventProcessor> myProducers;
	for (int i=0; i<numInputFragments; i++) {
		FileLoader prod(fileNames[i], fileWriter, dCounter);
		prod.ForkAndSpin();
		myProducers.Insert(prod);
	}

	// the thread running the main program is still available
	// we block it until the file writer dies
	fileWriter.WaitForProcessorDeath();

	DString::SaveDictionary();

	cout << "Load RELATION_NAME done : " << setprecision(6) << t.GetTime() <<
	" seconds" << endl;

	//kill the disk array
	DiskArray& instance = DiskArray::GetDiskArray();
	instance.Seppuku();

	delete dCounter;

  return 0;
}
